package com.xxx.rpc.server;
 
import com.xxx.rpc.common.bean.RpcRequest;
import com.xxx.rpc.common.bean.RpcResponse;
import com.xxx.rpc.common.util.Global;
import com.xxx.rpc.common.util.StringUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelHandler.Sharable;
import net.sf.cglib.reflect.FastClass;
import net.sf.cglib.reflect.FastMethod;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * RPC 服务端PIPELINE处理器（用于处理 RPC 请求）
 *
 * @author huangyong & Rosun
 * @since 1.0.0
 */
@Sharable
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServerHandler.class);

    /**
     * 存放 服务名 与 服务对象 之间的映射关系<br>
     * key: serviceName with version ; <br>
     * value : service bean object;<br>
     */
    private final Map<String, Object> handlerMap;
    
//    /**
//     * 是否需要调试和验证下requestid的唯一性？？debug用，生产环境请务必设置为false；
//     */
//    private boolean validate = true;
    
    /**
     * just for test...
     */
    //private Set<String> reqIdSet = new HashSet<String>();

    public RpcServerHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx,final RpcRequest request) throws Exception {
    	//服务端使用线程池来处理 应对并发！
    	Global.getThreadPool().execute(new Runnable() {
			@Override
			public void run() {
		    	LOGGER.debug("*** begin receive request:" + request.toString());
		        // 创建并初始化 RPC 响应对象
		        RpcResponse response = new RpcResponse();
		        response.setRequestId(request.getRequestId());
		        try {
		            Object result = handle(request);
		            response.setResult(result);
		        } catch (Exception e) {
		            LOGGER.error("handle result failure", e);
		            response.setException(e);
		        }
		        //写入 RPC 响应对象 不关闭连接；//by 罗珊
		        ctx.writeAndFlush(response).addListener(new io.netty.channel.ChannelFutureListener() {
                    @Override
                    public void operationComplete(io.netty.channel.ChannelFuture channelFuture) throws Exception {
                        LOGGER.debug("Send response for request " + request.toString());
                    }
                });
			}
		});//提交到线程池
    }

    private Object handle(RpcRequest request) throws Exception {
        // 获取服务对象
        String serviceName = request.getInterfaceName();
        String serviceVersion = request.getServiceVersion();
        if (StringUtil.isNotEmpty(serviceVersion)) {
            serviceName += "-" + serviceVersion;
        }
        Object serviceBean = handlerMap.get(serviceName);
        if (serviceBean == null) {
            throw new RuntimeException(String.format("can not find service bean by key: %s", serviceName));
        }
        // 获取反射调用所需的参数
        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();
        // 执行反射调用
//        Method method = serviceClass.getMethod(methodName, parameterTypes);
//        method.setAccessible(true);
//        return method.invoke(serviceBean, parameters);
        // 使用 CGLib 执行反射调用
        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
        return serviceFastMethod.invoke(serviceBean, parameters);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error("server caught exception.{},{}", cause.getMessage(),cause.getClass().getName());
        ctx.close();
        LOGGER.warn("server 已处理 客户端断开事件。");
    }
}
